#include "../../core/pch.h"
#include "../../pi/binarysemaphore.h"
#include "../utils/utils.h"

/* Information thread. Prints info table on
   standard output. */

static void* 
info_thread(void* args)
{
	/* Argument to logger_func containing data about connected client. */ 
	struct info_thread_arg info_arg = *((struct info_thread_arg*) args);
	struct info_table *shm 			= info_arg.shm;
	int sem_id 						= info_arg.sem_id;

	while (true)
	{
		DEBUGS("info thread");
		BinarySemaphore::Decrement(sem_id);
		IT_print(shm);
		BinarySemaphore::Increment(sem_id);
		sleep(WAIT_FOR_NEXT_PRINT_INFO_TABLE);
	}
}

/* Logger thread gathers information about
   connected clients. It checks if clients
   are still alive. */

static void* 
logger_thread(void* args)
{
	DEBUGS("logger: thread started");

	/* Argument to logger_func containing data about connected client. */ 
	struct logger_thread_arg logger_arg = *((struct logger_thread_arg*) args);

	/* Indicates if connected client is still alive. */
	bool isAlive = false;

	/* Get ip, port and pid of connected client. */
	char *ip 				= inet_ntoa(logger_arg.cliaddr.sin_addr);
	int port 				= logger_arg.cliaddr.sin_port;
	int pid  				= logger_arg.slave_pid;
	struct info_table *shm 	= logger_arg.shm;
	int sem_id				= logger_arg.sem_id;
	FILE *primes_file		= logger_arg.primes_file;

	/* Non blocking socket. */
	fcntl(logger_arg.client_sock, F_SETFL, O_NONBLOCK);

	/* Get current time of day. */
	struct timeval tv;
	gettimeofday(&tv, NULL);
	time_t startTime = tv.tv_sec;
	
	while (true)
	{
		gettimeofday(&tv, NULL);

		/** Sleep some time before next read. */
		sleep(TIME_TO_WAIT_FOR_DATA);

		char msg;
		if (recv(logger_arg.client_sock, &msg, sizeof(char), 0) != NO_DATA_TO_READ)
		{
			switch (msg)
			{
				case ALIVE_MSG:
					DEBUGS("ALIVE_MSG");
					isAlive = true;
					startTime = tv.tv_sec;
					break;
				case IM_IDLE:
					DEBUGS("make idle ip = %s, port = %d, pid = %d", ip, port, pid);
					BinarySemaphore::Decrement(sem_id);
					IT_make_idle(ip, port, pid, shm);
					BinarySemaphore::Increment(sem_id);
					break;
			}
			msg = NONE;
		}

		if (!isAlive && (tv.tv_sec - startTime) > TIMEOUT_FOR_IM_ALIVE)
		{
			BinarySemaphore::Decrement(sem_id);
			DEBUGS("logger: no results from %s", ip);
			IT_make_dead(ip, port, shm);
			PF_set_not_received_and_not_sent(IT_get_number(ip, port, shm), primes_file);
			BinarySemaphore::Increment(sem_id);
			pthread_exit(NULL);
		}

		isAlive = false;
	}
}

/* Initialiaze network settings for logger. */

static void
init_network(struct sockaddr_in &log_addr, int &log_sock, int log_port)
{
	FILL_SOCKADDR_IN_INT(log_addr, AF_INET, log_port, INADDR_ANY);
	log_sock = socket(AF_INET, SOCK_STREAM, 0); 
	if (log_sock == SOCKET_ERROR)
	{
		DEBUGS("logger: socket create error (%s)", strerror(errno));
		exit(SOCKET_ERROR);
	}

	/** Don't bother if socket was used. */
	int yes = 1;
	int ret = setsockopt(log_sock, SOL_SOCKET, SO_REUSEADDR, &yes, sizeof(int));
	if (ret == SETSOCKOPT_ERROR)
	{
		DEBUGS("logger: socket option error (%s)", strerror(errno));
		exit(SETSOCKOPT_ERROR);
	}

	ret = bind(log_sock, (struct sockaddr*) &log_addr, sizeof(log_addr)); 
	if (ret == BIND_ERROR)
	{
		DEBUGS("logger: bind error (%s)", strerror(errno));
		exit(BIND_ERROR);
	}

	ret = listen(log_sock, 64);
	if (ret == LISTEN_ERROR)
	{
		DEBUGS("logger: listen error (%s)", strerror(errno));
		exit(LISTEN_ERROR);
	}
}

/* Main logger function waits for connections.
   If slave connected it starts a gatherer thread
   for this slave. */

int 
logger(int segment_id, int sem_id, int logger_port, struct info_table *shm,
	FILE *primes_file)
{
	DEBUGS("logger: started");

	/* Number of runing threads. */
	int thread_count = 0;

	/* Network variables. */
	struct sockaddr_in logger_addr, cliaddr;
	int logger_sock;
	socklen_t len = sizeof(struct sockaddr);

	/* Initialize network. */
	init_network(logger_addr, logger_sock, logger_port);

	DEBUGS("logger: after init_network");

	/* Threads specific structures. */
	pthread_t logger_t[MAX_SLAVES];
	pthread_t info_t;
	struct logger_thread_arg logger_arg[MAX_SLAVES];
	pthread_attr_t attr;

	/* Make all threads joinable. */
	pthread_attr_init(&attr);
	pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_JOINABLE);

	/* Run info thread. */
	info_thread_arg info_arg;
	info_arg.sem_id = sem_id;
	info_arg.shm	= shm;
	pthread_create(&info_t, NULL, info_thread, (struct info_arg*) &info_arg);

	while (true)
	{
		DEBUGS("logger: waiting for connection");

		int client_sock = accept(logger_sock, (struct sockaddr*) &cliaddr, &len);

		DEBUGS("logger: accepted connection");

		pid_t slave_pid;
		int worker_port;
		recv(client_sock, &slave_pid, sizeof(pid_t), 0);
		recv(client_sock, &worker_port, sizeof(int), 0);

		DEBUGS("logger: slaves pid %d", slave_pid);

		/* Start thread for gathering slave information. */
		logger_arg[thread_count].client_sock 	= client_sock;
		logger_arg[thread_count].cliaddr 		= cliaddr;
		logger_arg[thread_count].slave_pid 		= slave_pid;
		logger_arg[thread_count].shm 			= shm;
		logger_arg[thread_count].sem_id			= sem_id;
		logger_arg[thread_count].primes_file	= primes_file;

		/* Add entry to info table. */
		BinarySemaphore::Decrement(sem_id);
		IT_add(inet_ntoa(cliaddr.sin_addr), cliaddr.sin_port, 
			slave_pid, worker_port, shm);
		BinarySemaphore::Increment(sem_id);

		/* Create thread. */
		pthread_create(&logger_t[thread_count], &attr, logger_thread, 
			(struct logger_thread_arg*) &logger_arg[thread_count]);

		thread_count++;
	}

	DEBUGS("logger: exited");

	pthread_exit(NULL);

	return 0;
}

